package cn.edu.flink.scala.tutorial.window

import cn.edu.flink.scala.tutorial.source.Event
import cn.edu.flink.scala.tutorial.window.AllowedLatenessTest.{EventsSource, WatermarkWindowResult}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

object SideOutputLateDataTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val events = List(
      Event("Mary", "home/", 1000L),
      Event("Mary", "cart/", 2000L),
      Event("Mary", "cart/", 3000L),
      Event("Mary", "cart/", 10000L),
      Event("Mary", "cart/", 6000L),
      Event("Mary", "cart/", 20000L),
      Event("Mary", "cart/", 30000L),
      Event("Mary", "cart/", 40000L))

    val eventsDS = env.addSource(new EventsSource(events))
      .assignAscendingTimestamps(_.timestamp)

    eventsDS.print("input")
    val outputTag = OutputTag[Event]("late-data")


    val result = eventsDS.keyBy(_.user)
      .window(TumblingEventTimeWindows.of(Time.seconds(10)))
      .sideOutputLateData(outputTag)
      .process(new WatermarkWindowResult)

    result.print("result")

    result.getSideOutput(outputTag).print("late-data")

    env.execute("AllowedLatenessTest")
  }
}
